/*
 * Copyright 2009 Armando Blancas
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package taskgraph.tasks.data;

import java.io.IOException;
import java.io.InterruptedIOException;

import taskgraph.ports.BooleanOutputPort;
import taskgraph.ports.ByteOutputPort;
import taskgraph.ports.DoubleOutputPort;
import taskgraph.ports.FloatOutputPort;
import taskgraph.ports.IntOutputPort;
import taskgraph.ports.LongOutputPort;
import taskgraph.ports.OutputPort;
import taskgraph.ports.ShortOutputPort;
import taskgraph.tasks.PrimitiveTask;


/**
 * Reads a native array, putting each element through the output port.
 * 
 * <p>The input array may be of any native type. The output port may be
 * the appropriate port for that native type, or an OutputPort<Object>
 * where each reference is a wrapper for a native type.
 * 
 * @author Armando Blancas
 */
public class ReadNativeArray extends PrimitiveTask {

	/*
	 * References to native arrays.
	 */
	private boolean[]booleanArray;
	private byte[] byteArray;
	private short[] shortArray;
	private int[] intArray;
	private long[] longArray;
	private float[] floatArray;
	private double[] doubleArray;
	
	/*
	 * References to type-specific output ports.
	 */
	private BooleanOutputPort booleanOutput;
	private ByteOutputPort byteOutput;
	private ShortOutputPort shortOutput;
	private IntOutputPort intOutput;
	private LongOutputPort longOutput;
	private FloatOutputPort floatOutput;
	private DoubleOutputPort doubleOutput;
	
	/*
	 * Generic output port for wrapper objects.
	 */
	private OutputPort<Object> output;
	
	/**
	 * This constructor allows creating instances as beans.
	 */
    public ReadNativeArray() {
    	// do nothing
    }
    
    /**
     * ReadNativeArray constructor.
     * 
     * @param booleanOutput An output port for transporting booleans
     * @param booleanArray A {@code boolean} array with source data.
     */
	public ReadNativeArray(BooleanOutputPort booleanOutput, 
			               boolean[] booleanArray) {
		setBooleanOutput(booleanOutput);
		setBooleanArray(booleanArray);
    }

    /**
     * ReadNativeArray constructor.
     * 
     * @param byteOutput An output port for transporting bytes.
     * @param byteArray A {@code byte} array with source data.
     */
	public ReadNativeArray(ByteOutputPort byteOutput, 
			               byte[] byteArray) {
		setByteOutput(byteOutput);
		setByteArray(byteArray);
    }

    /**
     * ReadNativeArray constructor.
     * 
     * @param doubleOutput An output port for transporting doubles.
     * @param doubleArray A {@code double} array with source data.
     */
	public ReadNativeArray(DoubleOutputPort doubleOutput, 
			               double[] doubleArray) {
		setDoubleOutput(doubleOutput);
		setDoubleArray(doubleArray);
    }

    /**
     * ReadNativeArray constructor.
     * 
     * @param floatOutput An output port for transporting floats.
     * @param floatArray A {@code float} array with source data.
     */
	public ReadNativeArray(FloatOutputPort floatOutput, 
			               float[] floatArray) {
		setFloatOutput(floatOutput);
		setFloatArray(floatArray);
    }

    /**
     * ReadNativeArray constructor.
     * 
     * @param intOutput An output port for transporting integers.
     * @param intArray An {@code int} array with source data.
     */
	public ReadNativeArray(IntOutputPort intOutput, 
			               int[] intArray) {
		setIntOutput(intOutput);
		setIntArray(intArray);
    }

    /**
     * ReadNativeArray constructor.
     * 
     * @param longOutput An output port for transporting long integers.
     * @param longArray A {@code long} array with source data.
     */
	public ReadNativeArray(LongOutputPort longOutput, 
			               long[] longArray) {
		setLongOutput(longOutput);
		setLongArray(longArray);
    }

    /**
     * ReadNativeArray constructor.
     * 
     * @param output An output port for transporting wrapper objects.
     */
	public ReadNativeArray(OutputPort<Object> output) {
		setOutput(output);
    }

    /**
     * ReadNativeArray constructor.
     * 
     * @param shortOutput An output port for transporting short integers.
     * @param shortArray A {@code short} array with source data.
     */
	public ReadNativeArray(ShortOutputPort shortOutput, 
			               short[] shortArray) {
		setShortOutput(shortOutput);
		setShortArray(shortArray);
    }

	private void close() {
		if (booleanOutput != null)
			booleanOutput.close();
		if (byteOutput != null)
			byteOutput.close();
		if (shortOutput != null)
			shortOutput.close();
		if (intOutput != null)
			intOutput.close();
		if (longOutput != null)
			longOutput.close();
		if (floatOutput != null)
			floatOutput.close();
		if (doubleOutput != null)
			doubleOutput.close();
		if (output != null)
			output.close();
	}

	/**
	 * Gets the reference to the given boolean array.
	 * 
	 * @return The array reference.
	 */
	public boolean[] getBooleanArray() {
		return booleanArray;
	}

	/**
	 * Gets the reference to the given boolean port.
     *
	 * @return The boolean output port.
	 */
	public BooleanOutputPort getBooleanOutput() {
		return booleanOutput;
	}

	/**
	 * Gets the reference to the given byte array.
	 * 
	 * @return The array reference.
	 */
	public byte[] getByteArray() {
		return byteArray;
	}

	/**
	 * Gets the reference to the given byte port.
	 * 
	 * @return The byte output port.
	 */
	public ByteOutputPort getByteOutput() {
		return byteOutput;
	}

	/**
	 * Gets the reference to the given double array.
	 * 
	 * @return The array reference.
	 */
	public double[] getDoubleArray() {
		return doubleArray;
	}

	/**
	 * Gets the reference to the given double port.
	 * 
	 * @return The double output port.
	 */
	public DoubleOutputPort getDoubleOutput() {
		return doubleOutput;
	}

	/**
	 * Gets the reference to the given float array.
	 * 
	 * @return The array reference.
	 */
	public float[] getFloatArray() {
		return floatArray;
	}

	/**
	 * Gets the reference to the given float port.
	 * 
	 * @return The float output port.
	 */
	public FloatOutputPort getFloatOutput() {
		return floatOutput;
	}

	/**
	 * Gets the reference to the given integer array.
	 * 
	 * @return The array reference.
	 */
	public int[] getIntArray() {
		return intArray;
	}

	/**
	 * Gets the reference to the given integer port.
	 * 
	 * @return The integer output port.
	 */
	public IntOutputPort getIntOutput() {
		return intOutput;
	}

	/**
	 * Gets the reference to the given long integer array.
	 * 
	 * @return The array reference.
	 */
	public long[] getLongArray() {
		return longArray;
	}

	/**
	 * Gets the reference to the given long integer port.
	 * 
	 * @return The long output port.
	 */
	public LongOutputPort getLongOutput() {
		return longOutput;
	}

	/**
     * Gets the output port, used for wrapper objects.
     * 
	 * @return The output port.
	 */
	public OutputPort<Object> getOutput() {
		return output;
	}

	/**
	 * Gets the reference to the given short integer array.
	 * 
	 * @return The array port.
	 */
	public short[] getShortArray() {
		return shortArray;
	}

	/**
	 * Gets the reference to the given short integer array.
	 * 
	 * @return the shortOutput
	 */
	public ShortOutputPort getShortOutput() {
		return shortOutput;
	}
	
	private void loopBoolean() 
	throws InterruptedIOException, ArrayIndexOutOfBoundsException, IOException {
		int i = 0;
		if (booleanOutput != null) {
			while (!Thread.currentThread().isInterrupted()) {
				booleanOutput.write(booleanArray[i++]);
			}
		} else if (output != null) {
			while (!Thread.currentThread().isInterrupted()) {
				output.write(booleanArray[i++]);
			}
		}
	}
	
	private void loopByte() 
	throws InterruptedIOException, ArrayIndexOutOfBoundsException, IOException {
		int i = 0;
		if (byteOutput != null) {
			while (!Thread.currentThread().isInterrupted()) {
				byteOutput.write(byteArray[i++]);
			}
		} else if (output != null) {
			while (!Thread.currentThread().isInterrupted()) {
				output.write(byteArray[i++]);
			}
		}
	}
	
	private void loopDouble() 
	throws InterruptedIOException, ArrayIndexOutOfBoundsException, IOException {
		int i = 0;
		if (doubleOutput != null) {
			while (!Thread.currentThread().isInterrupted()) {
				doubleOutput.write(doubleArray[i++]);
			}
		} else if (output != null) {
			while (!Thread.currentThread().isInterrupted()) {
				output.write(doubleArray[i++]);
			}
		}
	}
	
	private void loopFloat() 
	throws InterruptedIOException, ArrayIndexOutOfBoundsException, IOException {
		int i = 0;
		if (floatOutput != null) {
			while (!Thread.currentThread().isInterrupted()) {
				floatOutput.write(floatArray[i++]);
			}
		} else if (output != null) {
			while (!Thread.currentThread().isInterrupted()) {
				output.write(floatArray[i++]);
			}
		}
	}
	
	private void loopInt() 
	throws InterruptedIOException, ArrayIndexOutOfBoundsException, IOException {
		int i = 0;
		if (intOutput != null) {
			while (!Thread.currentThread().isInterrupted()) {
				intOutput.write(intArray[i++]);
			}
		} else if (output != null) {
			while (!Thread.currentThread().isInterrupted()) {
				output.write(intArray[i++]);
			}
		}
	}
	
	private void loopLong() 
	throws InterruptedIOException, ArrayIndexOutOfBoundsException, IOException {
		int i = 0;
		if (longOutput != null) {
			while (!Thread.currentThread().isInterrupted()) {
				longOutput.write(longArray[i++]);
			}
		} else if (output != null) {
			while (!Thread.currentThread().isInterrupted()) {
				output.write(longArray[i++]);
			}
		}
	}
	
	private void loopShort() 
	throws InterruptedIOException, ArrayIndexOutOfBoundsException, IOException {
		int i = 0;
		if (shortOutput != null) {
			while (!Thread.currentThread().isInterrupted()) {
				shortOutput.write(shortArray[i++]);
			}
		} else if (output != null) {
			while (!Thread.currentThread().isInterrupted()) {
				output.write(shortArray[i++]);
			}
		}
	}

	/**
     * Performs the work of this task.
     *
     * Dispatch the loop processing method according to the specified array.
     */
	@Override
    public void run() {
    	try {
    		if (booleanArray != null) {
    			loopBoolean();
    		} else if (byteArray != null) {
    			loopByte();
    		} else if (shortArray != null) {
    			loopShort();
    		} else if (intArray != null) {
    			loopInt();
    		} else if (longArray != null) {
    			loopLong();
    		} else if (floatArray != null) {
    			loopFloat();
    		} else if (doubleArray != null) {
    			loopDouble();
    		} else {
                log.error("ReadNativeArray#run(): array not set");    			
    		}
        } catch (InterruptedIOException e) {
        	// interruption request
        } catch (ArrayIndexOutOfBoundsException e) {
        	// we're done reading the array
        } catch (IOException e) {
        	// pipe synch error
            log.error("ReadNativeArray#run()", e);
		} finally {
			close();
        }
    }
	
	/**
	 * Sets a boolean array.
	 * 
	 * @param booleanArray The boolean array to set.
	 */
	public void setBooleanArray(boolean[] booleanArray) {
    	if (booleanArray == null)
    		throw new IllegalArgumentException("booleanArray == null");
		this.booleanArray = booleanArray;
	}

	/**
	 * Sets a boolean output port.
	 * 
	 * @param booleanOutput The boolean output port to set.
	 */
	public void setBooleanOutput(BooleanOutputPort booleanOutput) {
    	if (booleanOutput == null)
    		throw new IllegalArgumentException("booleanOutput == null");
		this.booleanOutput = booleanOutput;
	}

	/**
	 * Sets a byte array.
	 * 
	 * @param byteArray The byte array to set.
	 */
	public void setByteArray(byte[] byteArray) {
    	if (byteArray == null)
    		throw new IllegalArgumentException("byteArray == null");
		this.byteArray = byteArray;
	}

	/**
	 * Sets a byte output port.
	 * 
	 * @param byteOutput The byte output port to set.
	 */
	public void setByteOutput(ByteOutputPort byteOutput) {
    	if (byteOutput == null)
    		throw new IllegalArgumentException("byteOutput == null");
		this.byteOutput = byteOutput;
	}

	/**
	 * Sets a double array.
	 * 
	 * @param doubleArray The double array to set.
	 */
	public void setDoubleArray(double[] doubleArray) {
    	if (doubleArray == null)
    		throw new IllegalArgumentException("doubleArray == null");
		this.doubleArray = doubleArray;
	}

	/**
	 * Sets a double output port.
	 * 
	 * @param doubleOutput The double output port to set.
	 */
	public void setDoubleOutput(DoubleOutputPort doubleOutput) {
    	if (doubleOutput == null)
    		throw new IllegalArgumentException("doubleOutput == null");
		this.doubleOutput = doubleOutput;
	}

	/**
	 * Sets a float array.
	 * 
	 * @param floatArray The float array to set.
	 */
	public void setFloatArray(float[] floatArray) {
    	if (floatArray == null)
    		throw new IllegalArgumentException("floatArray == null");
		this.floatArray = floatArray;
	}

	/**
	 * Sets a float output port.
	 * 
	 * @param floatOutput The float output port to set.
	 */
	public void setFloatOutput(FloatOutputPort floatOutput) {
    	if (floatOutput == null)
    		throw new IllegalArgumentException("floatOutput == null");
		this.floatOutput = floatOutput;
	}

	/**
	 * Sets an integer array.
	 * 
	 * @param intArray The int array to set.
	 */
	public void setIntArray(int[] intArray) {
    	if (intArray == null)
    		throw new IllegalArgumentException("intArray == null");
		this.intArray = intArray;
	}

	/**
	 * Sets an integer output port.
	 * 
	 * @param intOutput The int output port to set.
	 */
	public void setIntOutput(IntOutputPort intOutput) {
    	if (intOutput == null)
    		throw new IllegalArgumentException("intOutput == null");
		this.intOutput = intOutput;
	}

	/**
	 * Sets a long integer array.
	 * 
	 * @param longArray The long array to set.
	 */
	public void setLongArray(long[] longArray) {
    	if (longArray == null)
    		throw new IllegalArgumentException("longArray == null");
		this.longArray = longArray;
	}


    /**
     * Sets a long integer output port.
     * 
	 * @param longOutput The long output port to set.
	 */
	public void setLongOutput(LongOutputPort longOutput) {
    	if (longOutput == null)
    		throw new IllegalArgumentException("longOutput == null");
		this.longOutput = longOutput;
	}

	/**
	 * Sets the output port, used for wrapper objects.
	 * 
	 * @param output The output port.
	 */
    public void setOutput(final OutputPort<Object> output) {
    	if (output == null)
    		throw new IllegalArgumentException("output == null");
 		this.output = output;
	}

	/**
	 * Sets a short array.
	 * 
	 * @param shortArray The short array to set.
	 */
	public void setShortArray(short[] shortArray) {
    	if (shortArray == null)
    		throw new IllegalArgumentException("shortArray == null");
		this.shortArray = shortArray;
	}

	/**
	 * Sets a short output port.
	 * 
	 * @param shortOutput The short output port to set.
	 */
	public void setShortOutput(ShortOutputPort shortOutput) {
    	if (shortOutput == null)
    		throw new IllegalArgumentException("shortOutput == null");
		this.shortOutput = shortOutput;
	}

}
